package com.ikas.ai.server.kafka;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONValidator;
import com.ikas.ai.enums.MeteTypeEnum;
import com.ikas.ai.model.KafkaMeteData;
import com.ikas.ai.server.module.data.model.doris.DorisYcData;
import com.ikas.ai.server.module.data.model.doris.DorisYxData;
import com.ikas.ai.server.module.data.service.DorisYcDataService;
import com.ikas.ai.server.module.data.service.DorisYxDataService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Timestamp;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 接收消息
 *
 * @author hubowei
 */
@Slf4j
@Component
public class KafkaDataConsumerService {

    @Autowired
    private DorisYxDataService dorisYxDataService;
    @Autowired
    private DorisYcDataService dorisYcDataService;


    @KafkaListener(topics = {KafkaInitialConfiguration.TOPIC_U_DATA}, topicPattern = "0", concurrency = "1",
            groupId = KafkaInitialConfiguration.TOPIC_U_DATA_GROUP,
            properties = {
                    "enable.auto.commit:false", "max.poll.interval.ms:60000"})
    @Transactional(rollbackFor = Exception.class)
    public void consumerUdata(List<ConsumerRecord<?, ?>> list, Acknowledgment ack) {
        if (updateData(list, KafkaInitialConfiguration.TOPIC_U_DATA)) {
            ack.acknowledge();
        }
    }


    @SneakyThrows
    @Transactional(rollbackFor = Exception.class)
    public boolean updateData(List<ConsumerRecord<?, ?>> list, String topic) {
        for (ConsumerRecord<?, ?> consumerRecord : list) {
            log.info("kafka数据组装之前:value:{},offset:{}", JSON.toJSONString(consumerRecord.value()), consumerRecord.offset());
        }
        //1.将kafka数据转换从实体对象
        List<KafkaMeteData> kafkaMeteDataList = coverRecordToMeteDataEntity(list);
        //2.将kafka数据存入doris相关表中
        handleKafkaMeteDataToDoris(kafkaMeteDataList, topic);
        return true;
    }

    /**
     * 将kafka数据转换从实体对象
     *
     * @param list
     * @return
     */
    private List<KafkaMeteData> coverRecordToMeteDataEntity(List<ConsumerRecord<?, ?>> list) {
        List<KafkaMeteData> meteDataList = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : list) {
            final Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                final String message = (String) kafkaMessage.get();
                // 兼容kafka数据两种json
                final JSONValidator.Type type = JSONValidator.from(message).getType();
                if (type == JSONValidator.Type.Array) {
                    meteDataList.addAll(JSONArray.parseArray(message, KafkaMeteData.class));
                } else if (type == JSONValidator.Type.Object) {
                    meteDataList.add(JSONObject.parseObject(message, KafkaMeteData.class));
                } else {
                    continue;
                }
            }
            log.info("====DATA-KAFKA-DORIS====>offset>[{}]--timestamp>[{}]------>meteDataList:{}", record.offset()
                    , record.timestamp(), JSONUtil.toJsonStr(meteDataList));
        }
        return meteDataList;
    }


    /**
     * 将kafka数据存入doris相关表中
     *
     * @param topic
     * @param kafkaMeteDatas
     */
    private void handleKafkaMeteDataToDoris(List<KafkaMeteData> kafkaMeteDatas, String topic) {
        List<DorisYxData> yxDataList = new ArrayList<>();
        List<DorisYcData> ycDataList = new ArrayList<>();
        for (KafkaMeteData kafkaMeteData : kafkaMeteDatas) {
            if (StringUtils.isBlank(kafkaMeteData.getMeteCode())) {
                log.warn("server-kafka-t-data consumer：[warn] meteCode is null. topic = {} {}.",
                        topic, JSON.toJSONString(kafkaMeteData));
                continue;
            }
            if (StringUtils.isBlank(kafkaMeteData.getMachineNo())) {
                kafkaMeteData.setMachineNo(topic);
            }
            final Integer meteType = kafkaMeteData.getMeteType();
            if (MeteTypeEnum.YX.getNum().equals(meteType)) {
                DorisYxData dorisYxData = BeanUtil.copyProperties(kafkaMeteData, DorisYxData.class);
                dorisYxData.setModelUse(kafkaMeteData.getModelUse());
                dorisYxData.setReportTime(new Timestamp(kafkaMeteData.getGenerateTime()));
                dorisYxData.setUpdateTime(new Date());
                dorisYxData.setCreateTime(new Date());
                yxDataList.add(dorisYxData);
            }
            if (MeteTypeEnum.YC.getNum().equals(meteType)) {
                DorisYcData dorisYcData = BeanUtil.copyProperties(kafkaMeteData, DorisYcData.class);
                dorisYcData.setModelUse(kafkaMeteData.getModelUse());
                dorisYcData.setReportTime(new Timestamp(kafkaMeteData.getGenerateTime()));
                dorisYcData.setUpdateTime(new Date());
                dorisYcData.setCreateTime(new Date());
                ycDataList.add(dorisYcData);
            }
        }
        if (!CollectionUtils.isEmpty(yxDataList)) {
            log.info("dorisYxList:{}", JSON.toJSONString(yxDataList));
            dorisYxDataService.saveBatch(yxDataList);
        }
        if (!CollectionUtils.isEmpty(ycDataList)) {
            log.info("dorisYcList:{}", JSON.toJSONString(ycDataList));
            dorisYcDataService.saveBatch(ycDataList);
        }
    }

}



















